package org.rui.rpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import org.rui.rpc.config.RegistryConfig;
import org.rui.rpc.model.ServiceMetaInfo;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author: linweirui
 * @date: 2024/12/04 14:09
 */

/**
 * etcd注册中心实现类
 */
public class EtcdRegistry implements Registry {
    private Client client;

    private KV kvClient;
    /**
     * 本机注册的节点 key 集合（用于维护续期）,这里只是存地址key，获取不了服务
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();
    /**
     * 注册中心服务缓存，这里可以获取服务全部注册节点信息
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();
    /**
     * 正在监听的 key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();


    /**
     * 根节点
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";

    /**
     * etcd注册中心初始化
     *
     * @param registryConfig
     */
    @Override
    public void init(RegistryConfig registryConfig) {
        //构造客户端
        client = Client.builder()
                .endpoints(registryConfig.getAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        //用于对 etcd 中的键值对进行操作。
        // 通过 kvClient 可以进行设置值、获取值、删除值、列出目录等操作。
        kvClient = client.getKVClient();
        heartBeat();
    }

    /**
     * 服务注册在第三方中间件上
     *
     * @param serviceMetaInfo 服务元数据
     * @throws Exception
     */
    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        //创建lease客户端和kv客户端
        Lease leaseClient = client.getLeaseClient();
        //设置30s租约
        long leaseId = leaseClient.grant(30).get().getID();
        //设置缓存的键值对
        String registryKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registryKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);
        //将键值对与租约关系关联起来，设置过期时间
        PutOption putOption = PutOption.builder()
                .withLeaseId(leaseId)
                .build();
        kvClient.put(key, value, putOption).get();
        //添加注册信息到本地缓存
        localRegisterNodeKeySet.add(registryKey);
    }

    /**
     * 服务注销
     *
     * @param serviceMetaInfo
     */
    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String registryKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(registryKey,
                StandardCharsets.UTF_8));
        localRegisterNodeKeySet.remove(registryKey);
    }

    /**
     * 服务发现
     *
     * @param serviceKey 服务键名
     * @return
     */
    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        List<ServiceMetaInfo> serviceMetaInfos = registryServiceCache.readCache();
        //如果存在则直接读缓存返回
        if (!CollUtil.isEmpty(serviceMetaInfos)) {
            return serviceMetaInfos;
        }
        //前缀搜索，结尾要加 /
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";
        //前缀查询
        GetOption getOption = GetOption.builder()
                .isPrefix(true)
                .build();
        try {
            //这里的key、value结构是前缀搜索的结果
            // key比如为searchPrefix+localhost：8080等 value为ServiceMetaInfo的json字符串
            List<KeyValue> keyValues = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8), getOption)
                    .get()
                    .getKvs();
            //解析服务信息
            List<ServiceMetaInfo> serviceMetaInfoList = keyValues.stream()
                    .map(keyValue -> {
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                        // 监听 每一个具体实现类服务信息 key 的变化
                        watch(key);
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    }).collect(Collectors.toList());
            if (CollUtil.isEmpty(serviceMetaInfoList)) {
                //集合为空，说明对应的服务发现节点信息为空
                return Collections.emptyList();
            }
            //写入缓存
            registryServiceCache.writeCache(serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败", e);
        }
    }

    /**
     * 销毁服务
     */
    @Override
    public void destroy() {
        System.out.println("===========当前节点下线==============");
        for (String key : localRegisterNodeKeySet) {
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(key + " 节点下线失败 ：", e);
            }
        }
        // 释放资源
        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }

    }

    /**
     * 服务端心跳检测
     */
    @Override
    public void heartBeat() {
        //每10s重新注册续签一下
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                for (String key : localRegisterNodeKeySet) {
                    try {
                        //获取某个节点的key和对应的value（ServiceMetaInfo）信息
                        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8))
                                .get()
                                .getKvs();
                        //如果获取的节点信息为空，那么说明节点过期，跳过
                        if (CollUtil.isEmpty(keyValues)) {
                            continue;
                        }
                        //节点不为空，进行续期，即重新注册操作
                        KeyValue keyValue = keyValues.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key + " 续签失败", e);
                    }
                }
            }
        });
        // 支持秒级别定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();

    }

    /**
     * 消费端服务缓存更新监听
     *
     * @param serviceNodeKey
     */
    @Override
    public void watch(String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        if (newWatch) {
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                for (WatchEvent event : response.getEvents()) {
                    switch (event.getEventType()) {
                        //删除时触发
                        case DELETE:
                            //清理掉注册服务缓存信息，这里的缓存单纯存的是对应实现类的全部服务节点信息
                            registryServiceCache.clearCache();
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }
    /*public static void main(String[] args) throws ExecutionException, InterruptedException {
        Client client = Client.builder()
                .endpoints("http://localhost:2379")
                .build();
        KV kvClient = client.getKVClient();
        ByteSequence key = ByteSequence.from("test_key".getBytes());
        ByteSequence value = ByteSequence.from("test_value".getBytes());
        kvClient.put(key,value).get();
        CompletableFuture<GetResponse> getFuture = kvClient.get(key);

        GetResponse response = getFuture.get();
        kvClient.delete(key).get();
    }*/

}
